package com.tm.test;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;

import javax.xml.crypto.Data;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @author Rocky
 * @create 2022-11-29 10:17
 */
public class RocketMQyanshi {
    public static void main23(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        //1.创建一个producer对象
        //消息查询时是根据Topic，tag来进行查询，生产者组实际上并没有什么意义
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("zkj_producer");
        //2.设置连接地址
        defaultMQProducer.setNamesrvAddr("192.168.0.3:9876");
        //3.生产者需要先启动
        defaultMQProducer.start();
        //4.设置消息
        String aaa="{name:'Jrodan',age:36}";
        //5.创建消息对象
        Message message = new Message("zkjnba","NBAtag",aaa.getBytes());

        //设置消息的延时等级
        //1 2 3 4 5 6 7 8 9 10  11  12  13  14  15  16  17  18
        //1s    5s  10s 30s 1m  2m  3m  4m  5m  6m  7m  8m  9m  10m 20m 30m 1h  2h
        message.setDelayTimeLevel(1);
        SendResult send = defaultMQProducer.send(message);
        System.out.println(send);
        defaultMQProducer.shutdown();
    }

    //延时消费
    public static void main5(String[] args) throws Exception {
        //创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zkj1");
        //设置nameServer
        consumer.setNamesrvAddr("192.168.0.3:9876");
        //设置消费消息的顺序
        //有现成的枚举,LAST就是先进先出
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置要读取的消息的topic和tag
        consumer.subscribe("NBA",MessageSelector.bySql("isMember = 1 AND isBitch = 1"));
        //顺序读取消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //设置下个消息接收的延时等级 消费间隔
                consumeConcurrentlyContext.setDelayLevelWhenNextConsume(3);
                //获取每条消息的延时时间
                for (MessageExt messageExt : list) {
                    byte[] body = messageExt.getBody();
                    String str = new String(body);
                    System.out.println(str);
                    long storeTimestamp = messageExt.getStoreTimestamp();
                    System.out.println("存储时间为：" + storeTimestamp);
                    System.out.println(body);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动消费者
        consumer.start();
        System.out.println("Consumer启动成功！");

    }



    public static void maindfs(String[] args) {
        Date date = new Date(1669689079135l);
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println(simpleDateFormat.format(date));
    }

    public static void main77(String[] args) throws Exception {
        //创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("kjl11");
        //设置nameServer
        consumer.setNamesrvAddr("192.168.0.3:9876");
        //设置消费消息的顺序
        //有现成的枚举,LAST就是先进先出
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置要读取的消息的topic和tag
        consumer.subscribe("jps",MessageSelector.bySql("isCool = 6 AND isMember=1"));
        //顺序读取消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                //设置消息读取后自动删除
                consumeOrderlyContext.setAutoCommit(true);
                //list中就是读取到的所有消息
                for (MessageExt messageExt : list) {
                    String msgId = messageExt.getMsgId();
                    String string = new String(messageExt.getBody());
                    JSONObject jsonObject = JSONObject.parseObject(string);
                    System.out.println(msgId);
                    System.out.println(string);
                    System.out.println(jsonObject);
                }
                //返回成功标识
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //启动消费者
        consumer.start();
        System.out.println("Consumer启动成功！");
    }


}
